package com.socket.batch.deal;

import com.socket.batch.transmit.ITransmitter;
import com.socket.batch.transmit.TransmitterFactory;
import com.socket.context.SpringContextTool;
import com.socket.entity.GpsRecord;
import com.socket.service.GpsBatchService;
import com.socket.service.GpsRecordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.web.context.ContextLoader;
import org.springframework.web.context.WebApplicationContext;

import java.util.*;
import java.util.concurrent.*;

import static com.socket.batch.transmit.TransmitterFactory.getTransmitter;

/**
 * Created by kaddddd on 2017/7/5 0005.
 */
public abstract class AbstractDeal implements IDeal{
    private static final Logger logger = LoggerFactory.getLogger(AbstractDeal.class);

    public static Map<String,BlockingQueue<GpsRecord>> mapQueue = new ConcurrentHashMap<String,BlockingQueue<GpsRecord>>();
    /**
     * 内存队列大小
     */
    private int memQueueSize = 3000;
    /**
     * 每次处理量
     */
    protected int eachQuantity = 80;
    /**
     * 处理频率
     */
    private int frequency = 1000;
    /**
     * 处理线程个数
     */
    private int threadNum = 2;

    private int threadIndex =0;
    /**
     * 批量操作中缓存数据的队列
     */
    private LinkedBlockingQueue queue = new LinkedBlockingQueue(3000);
    /**
     * 批量操作中使用的线程池
     */
    private ExecutorService executorService;

    private boolean start = false;

    public AbstractDeal() {
        // 创建线程池，并发执行SlaveThread
        executorService = Executors.newFixedThreadPool(threadNum, new ThreadFactory(){
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("batch-deal"+(threadIndex++));
                t.setDaemon(true);
                return t;
            }
        });
    }

    private BlockingQueue<GpsRecord> getQueue(String handlerName){
        BlockingQueue<GpsRecord> queue ;
        if(mapQueue.containsKey(handlerName)){
            return mapQueue.get(handlerName);
        }else{
            synchronized (mapQueue){
                queue = mapQueue.get(handlerName);
                if(queue==null){
                    queue = new LinkedBlockingQueue<GpsRecord>(memQueueSize);
                    mapQueue.put(handlerName,queue);
                }
            }
        }

        return queue;
    }

    public void deal(GpsRecord gpsRecord){
        if (!start) {
            start();
        }
        Set<String> handlerNames = new HashSet();
        handlerNames.add("socket");
        handlerNames.add("persist");

        for (String handlerName : handlerNames){
            //针对不同的处理器，获取不同的队列
            BlockingQueue<GpsRecord> queue = getQueue(handlerName);
            //如果经纬度为0,排除
            if (gpsRecord.getLat()!=0 && gpsRecord.getLon()!=0){
                //ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:applicationContext.xml");
                //程序启动后为啥不加载使用的xml配置，因为没有web.xml，所以没有加载，使用@Component
                ApplicationContext ctx = SpringContextTool.getApplicationContext();

                GpsRecordService gpsRecordService = (GpsRecordService)ctx.getBean("gpsRecordService");
                int onlineFlag = gpsRecordService.findOnlineFlag(gpsRecord.getVehicleNo());
                gpsRecord.setOnline(onlineFlag);
                boolean success = queue.offer(gpsRecord);
                if(!success){
                    //如果队列满了。直接写数据库
                    ITransmitter transmitter = null;
                    if(handlerName.equals("persist")){
                        try{
                            //根据名称获取，是处理消息推送，还是持久化
                            getTransmitter("default").transition(gpsRecord);
                        }catch (Throwable e){
                            //TODO 发生异常了,需要考虑日志数据的重发问题
                            if(transmitter!=null){
                                //如果处理器不空时
                            }
                            e.printStackTrace();
                            logger.error("处理单条数据报错!");
                        }
                    }

                }
            }

        }
    }

    class QueueConsumer implements Runnable{

        @Override
        public void run() {
            System.out.println("消费线程处理gps数据持久化!");
            while (!Thread.currentThread().isInterrupted()) {
                if (!Thread.currentThread().isInterrupted()) {
                    try {
                        //TODO 做成配置
                        Thread.sleep(frequency);
                    } catch (InterruptedException e) {
                    }
                }


                Set<ConcurrentHashMap.Entry<String, BlockingQueue<GpsRecord>>> entries
                        = mapQueue.entrySet();

                int handNum = 0;
                for (ConcurrentHashMap.Entry<String, BlockingQueue<GpsRecord>> entry : entries) {
                    if (entry.getKey().equals("persist")) {
                        BlockingQueue<GpsRecord> queue = entry.getValue();
                        //logger.info("持久化队列:"+queue);
                        System.out.println("持久化队列:" + queue);
                        if (queue.isEmpty()) {
                            continue;
                        }
                        List<GpsRecord> dataList = new ArrayList<GpsRecord>();
                        try {
                            dataList = batchDataList(queue);
                            handNum = dataList.size();
                        } catch (Exception e) {
                            logger.error("从队列中获取批量数据异常!");
                        }
                        if (!dataList.isEmpty()) {
                            ITransmitter transmitter = null;
                            try {
                                transmitter = getTransmitter("persist");
                                if (transmitter != null) {
                                    //logger.info("开始执行批量插入:"+dataList);
                                    System.out.println("开始执行批量插入:" + dataList);
                                    transmitter.transition(dataList);
                                }
                            } catch (Throwable e) {
                                e.printStackTrace();
                            }
                        }
                    }

                }
            }

        }
    }

    abstract List batchDataList(BlockingQueue queue);

    public synchronized void start() {
        System.out.println("开启线程处理gps数据持久化!");
        if(start){
            return;
        }
        start = true;
        for (int i = 0; i < threadNum; i++) {
            executorService.execute(new QueueConsumer());
        }
    }

    /**
     * 任务结束函数.
     */
    //@PreDestroy
    public void stop(){
        try {
            System.out.println("gps数据持久化任务结束!");
            executorService.shutdownNow();
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                System.err.println("Pool did not terminate");
            }
            mapQueue.clear();
            mapQueue = null;
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
        }
    }
}



